package cn.itcast.flink.transformation;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class FlatMapDemo {
    public static void main(String[] args) throws Exception {
        /**
         * 开发步骤：
         * 构建批处理运行环境
         * 构建本地集合数据源
         * 使用flatMap将一条数据经过处理转换为三条数据
         * 使用逗号分隔字段
         * 分别构建三条数据
         * 打印输出
         */

        //TODO 构建批处理运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //TODO 构建本地集合数据源
        DataStream<String> lines = env.readTextFile("./data/input/flatmap.log");

        //TODO 使用flatMap将一条数据经过处理转换为三条数据
        //TODO 使用逗号分隔字段
        //TODO 分别构建三条数据
        SingleOutputStreamOperator<String> result = lines.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String line, Collector<String> collector) throws Exception {
                String[] elements = line.split(",");
                collector.collect(elements[0] + "有" + elements[1]);
                collector.collect(elements[0] + "有" + elements[2]);
                collector.collect(elements[0] + "有" + elements[3]);
            }
        });

        //TODO 打印输出
        result.print();

        env.execute();
    }
}